package cas.ihep.hadoop.util;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.ByteStreams;
import org.luaj.vm2.LuaTable;
import org.luaj.vm2.LuaValue;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.*;
import java.util.concurrent.Callable;

public class HadoopRunConfig {
    private LuaTable table;
    private String pwd,logdir,sh;

    HadoopRunConfig(){
        this(null);
    }

    HadoopRunConfig(LuaTable tb){
        table=tb;
        pwd=System.getProperty("user.dir");
        logdir=System.getProperty("java.io.tmpdir");
        sh=System.getenv("SHELL");
    }

    public String workingDirectory(){
        if(table!=null) {
            LuaValue val = table.get("dir");
            if (val.isstring()) {
                return val.tojstring();
            }
        }
        return pwd;
    }

    public String shell(){
        if(table!=null) {
            LuaValue val = table.get("shell");
            if (val.isstring()) {
                return val.tojstring();
            }
        }
        return sh;
    }

    public List<String> sources(){
        if(table!=null) {
            LuaValue val = table.get("sources");
            if (val.istable()) {
                LuaTable tb = (LuaTable) val;
                int sz = tb.keyCount();
                ImmutableList.Builder<String> builder=ImmutableList.builder();
                for (int i = 0; i < sz; i++) {
                    builder.add(tb.get(i + 1).tojstring());
                }
                return builder.build();
            }
        }
        return Collections.emptyList();
    }

    public String logDirectory(){
        if(table!=null) {
            LuaValue val = table.get("logdir");
            if (val.isstring()) {
                return val.tojstring();
            }
        }
        return logdir;
    }

    public boolean logInherited(){
        if(table!=null){
            LuaValue ret=table.get("loginherited");
            if(ret.isboolean()){
                return ret.toboolean();
            }
        }
        return false;
    }

    public List<String> logs(){
        if(table!=null) {
            LuaValue val = table.get("logs");
            if (val.istable()) {
                ImmutableList.Builder<String> builder=ImmutableList.builder();
                LuaTable tb = (LuaTable) val;
                for (int i = 0; i < tb.keyCount(); i++) {
                    builder.add(tb.get(i + 1).tojstring());
                }
                return builder.build();
            }
        }
        return Collections.emptyList();
    }

    public void updateEnv(Map<String,String> env){
        if(table!=null){
            LuaValue val=table.get("env");
            if(val.istable()){
                LuaTable tb=(LuaTable)val;
                LuaValue keys[]=tb.keys();
                for (LuaValue key : keys) {
                    LuaValue str = tb.get(key);
                    if (key.isstring() && str.isstring()) {
                        env.put(key.tojstring(), str.tojstring());
                    }
                }
            }
        }
    }

    public Map<String,String> env(){
        if(table!=null){
            LuaValue val=table.get("env");
            if(val.istable()){
                ImmutableMap.Builder<String,String> builder=ImmutableMap.builder();
                LuaTable tb=(LuaTable)val;
                LuaValue keys[]=tb.keys();
                for(LuaValue key:keys){
                    LuaValue str=tb.get(key);
                    if(key.isstring() && str.isstring()){
                        builder.put(key.tojstring(),str.tojstring());
                    }
                }
                return builder.build();
            }
        }
        return Collections.emptyMap();
    }

    public static Callable<Integer> newTask(final InputStream is,final OutputStream os){
        return new Callable<Integer>() {
            @Override
            public Integer call() throws Exception{
                ByteStreams.copy(is,os);
                return 0;
            }
        };
    }
}
